package com.godme.concurrent.pattern;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.function.Function;

public class ThreadPerMessagePattern {


    static class ThreadFactory {
        static class TemplateThread<F, T> extends Thread {
            private final BlockingQueue<F> from;
            private final BlockingQueue<T> to;
            private final Function<F, T> func;

            public TemplateThread(String name, BlockingQueue<F> from, BlockingQueue<T> to, Function<F, T> func) {
                super(name);
                this.from = from;
                this.to = to;
                this.func = func;
            }


            @Override
            public void run() {
                try {
                    while (true) {
                        to.offer(func.apply(from.take()));
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        }


        public static <F, T> Thread  newThread(String name, BlockingQueue<F> from, BlockingQueue<T> to, Function<F, T> apply) {
            return new TemplateThread<F, T>(name, from, to, apply);
        }
    }


    public static void main(String[] args) {
        Function<Integer, Integer> plusOne = (e) -> e * 2;
        Function<Integer, String> toString = (e) -> {
            StringBuilder sb = new StringBuilder();
            for (int i = 0; i < e; i++) {
                sb.append("*");
            }
            return sb.toString();
        };

        LinkedBlockingDeque<Integer> request = new LinkedBlockingDeque<>();
        LinkedBlockingDeque<Integer> processor = new LinkedBlockingDeque<>();
        LinkedBlockingDeque<String> response = new LinkedBlockingDeque<>();

        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                request.offer(i);
            }
        }).start();

        new Thread(()->{
            while(true){
                try {
                    System.err.println(response.take());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();

        for (int i = 0; i < 2; i++) {
            ThreadFactory.newThread("processor-"+i, request, processor, plusOne).start();
        }

        for (int i = 0; i < 2; i++) {
            ThreadFactory.newThread("receive-"+i, processor, response, toString).start();
        }

    }


}
